package cn.doit.bean

import java.util.Date

import cn.doit.utils.TimeUtils
import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD

import scala.beans.BeanProperty

/**
  * Created by Administrator on 2018/11/16.
  */
class LogBean extends Serializable {

  //最后访问时间
  @BeanProperty
  var lastTime = "defalult_duoyi"

  @BeanProperty
  var country = "defalult_duoyi"

  @BeanProperty
  var session = "defalult_duoyi"

  @BeanProperty
  var city_code = "defalult_duoyi"

  @BeanProperty
  var province_code = "defalult_duoyi"

  @BeanProperty
  var browserType = "defalult_duoyi"

  @BeanProperty
  var uid = "defalult_duoyi"

  @BeanProperty
  var province = "defalult_duoyi"

  @BeanProperty
  var browser = "defalult_duoyi"

  @BeanProperty
  var timestamp = "defalult_duoyi"

  @BeanProperty
  var deviceType = "defalult_duoyi"

  @BeanProperty
  var os = "defalult_duoyi"

  @BeanProperty
  var url = "defalult_duoyi"

  @BeanProperty
  var country_code = "defalult_duoyi"

  @BeanProperty
  var newUser = "defalult_duoyi"

  @BeanProperty
  var siteId = "defalult_duoyi"

  @BeanProperty
  var eventType = "defalult_duoyi"

  @BeanProperty
  var day = "defalult_duoyi"

  @BeanProperty
  var hour = "defalult_duoyi"

  @BeanProperty
  var minute = "defalult_duoyi"
  /**
    * 来源
    */
  @BeanProperty
  var source = "defalult_duoyi"

}

object LogBean {

  def parseRdd2LogBeanRdd(rdd: RDD[ConsumerRecord[String, String]]): RDD[LogBean] = {

    val partitions: RDD[LogBean] = rdd.mapPartitions(mp => {
      val map: Iterator[LogBean] = mp.map(line => {
        val lineJson = line.value()
        var logBean = new LogBean
        logBean = JSON.parseObject(lineJson, classOf[LogBean])
        //获取访问时间戳
        val ts = logBean.getTimestamp
        //        val ts = new Date().getTime.toString

        logBean.setDay(TimeUtils.dateFormatForYMD(ts))
        logBean.setHour(TimeUtils.dateFormatForHour(ts))
        logBean.setMinute(TimeUtils.dateFormatForMinute(ts))

//        logBean.setSource((Math.random() * 10 % 6).toInt.toString)
        logBean.setSource("0")

        logBean
      })
      map
    })
    partitions
  }

  def apply(): LogBean = new LogBean()
}
